package com.vincent.leader;

import com.vincent.common.*;
import com.vincent.common.Process;
import com.vincent.message.AdoptedMessage;
import com.vincent.message.PreemptedMessage;
import com.vincent.message.ProposeMessage;
import com.vincent.utils.BallotNumber;
import com.vincent.utils.Config;
import com.vincent.utils.PValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Leader接收replica的request消息，序列化(serialize)该request并回复消息给replica
 *  Leader维持四个变量：
 *
 *  1、ballot_number：一个单调递增的选票号
 *  2、active：一个布尔变量，初始化为false
 *  3、proposals：一个slot号到提议的Command，初始化为空，对应每个slot至多一个Command
 *  4、timeout：Leader在操作中等待的超时
 *
 * Created by wensen on 2016/12/11.
 */
public class Leader extends Process{

    private static Logger LOGGER = LoggerFactory.getLogger(Leader.class);

    private BallotNumber ballot_number;
    private boolean active;
    private Map<Long, Command> proposals;
    private double timeout;
    private Config config;

    public Leader(Environment env, int id, Config config) {
        super(env, id);
        this.ballot_number = new BallotNumber(0, this.id());
        this.active = false;
        this.proposals = new HashMap<Long, Command>();
        this.timeout = 1.0;
        this.config = config;
        this.addProc(this);
    }

    /**
     * leader开始时使用scout(spawning a scout)来初始化选票, 然后进入一个循环等待消息
     * 这里有三种消息将触发状态转移：
     *
     * 1、Propose：一个replica提议给定slot的命令(command)
     * 2、Adopted：由scout发送，表示当前的ballot被过半acceptor所接受
     *      (如果一个已经采用(adopted)的消息到来时带着旧的ballot number,则忽略它)
     *      pvalues包含由所有acceptor所采用的ballot number
     * 3、Preedpted：由scout或aommander发送，表示有更高ballot number被acceptor所采用
     *      如果该ballot number大于当前leader的ballot number,那么意味着不会使用当前
     *      leader的slot number选择command
     */
    public void body() {

        LOGGER.debug("Leader: Here I am: {}", this.id());
        //Scout发送p1a信息
        new Scout(this.env(), this.id(), this.id(), this.config.acceptors(), this.ballot_number);

        while (true) {

            Message msg = null;
            try {
                msg = this.getNextMessage();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            if (msg instanceof ProposeMessage) {

                LOGGER.debug("Leader: recved ProposeMessage");

                if (!this.proposals.containsKey(((ProposeMessage) msg).slot_number())) {
                    this.proposals.put(((ProposeMessage) msg).slot_number(),
                            ((ProposeMessage) msg).command());
                    if (this.active) {
                        new Commander.Builder().env(this.env()).id(this.id()).leader(this.id())
                                .acceptors(this.config.acceptors()).replicas(this.config.replicas())
                                .ballot_number(this.ballot_number).slot_number(((ProposeMessage) msg).slot_number())
                                .command(((ProposeMessage) msg).command()).build();
                    }
                }

            } else if (msg instanceof AdoptedMessage) {

                LOGGER.debug("Leader: recved AdoptedMessage");

                if (this.timeout > Constant.TIMEOUTSUBTRACT) {
                    this.timeout -= Constant.TIMEOUTSUBTRACT;
                    LOGGER.debug("Timeout decresed: {}", this.timeout);
                }

                if (this.ballot_number == ((AdoptedMessage) msg).ballot_number()) {

                    Map<Long, BallotNumber> pmax = new HashMap<Long, BallotNumber>();
                    //
                    //对于每一个slot number，将带有最高协议号的proposal放入proposals中
                    Map<Long, PValue> pvalues = ((AdoptedMessage) msg).accepted().pvalues();
                    for (Map.Entry<Long, PValue> pv : pvalues.entrySet()) {
                        if (!pmax.containsKey(pv.getKey()) ||
                                pmax.get(pv.getKey()).compareTo(pv.getValue().ballot_number()) < 0) {
                            pmax.put(pv.getKey(), pv.getValue().ballot_number());
                            this.proposals.put(pv.getKey(), pv.getValue().command());
                        }
                    }
                    //对于每一个proposal开始commander(第2阶段)
                    for (Map.Entry<Long, Command> sn : this.proposals.entrySet()) {
                        new Commander.Builder().env(this.env()).id(this.id()).leader(this.id())
                                .acceptors(this.config.acceptors()).replicas(this.config.replicas())
                                .ballot_number(this.ballot_number).slot_number(sn.getKey())
                                .command(sn.getValue()).build();
                    }
                    this.active = true;
                }

            } else if (msg instanceof PreemptedMessage) {

                LOGGER.debug("Leader: recved PreemptedMessage");

                //本leader与其它leader竞争
                if (((PreemptedMessage) msg).ballot_number().leader_id() > this.id()) {
                    this.timeout *= Constant.TIMEOUTMULTIPLY;
                    LOGGER.debug("{}, Timeout increased: {}", this.id(), this.timeout);
                }

                if (((PreemptedMessage) msg).ballot_number().compareTo(this.ballot_number) > 0) {
                    this.ballot_number = new BallotNumber(((PreemptedMessage) msg).ballot_number()
                                .round()+1, this.id());
                    new Scout(this.env(), this.id(), this.id(), this.config.acceptors(), this.ballot_number);
                }
                this.active = false;

            } else {

                LOGGER.debug("Leader: unknown msg type");
            }

            //睡眠timeout秒
            try {
                TimeUnit.MILLISECONDS.sleep((long) this.timeout*1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }
}
